package com.zy.demo;

import com.zy.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.Test;

import java.time.LocalDateTime;
import java.util.List;

/**
 * 延迟消息
 */
public class MsTest {

    @Test
    public void testDelay() throws Exception {

        DefaultMQProducer producer = new DefaultMQProducer("ms-produce-group");

        producer.setNamesrvAddr(MqConstant.NAMESRV_ADDR);

        producer.start();

        Message message = new Message("msTopic", "ms msg".getBytes());

        //给消息设置延迟时间，官网查询等级，可以自己配置rocket broker的配置文件
        message.setDelayTimeLevel(3);

        producer.send(message);

        System.out.println("send message date:" + LocalDateTime.now());

        producer.shutdown();

    }



    /**
     *
     * 2024-05-15T10:38:22.292
     * 2024-05-15T10:38:41.471
     * ================================================
     * 2024-05-15T10:39:12.218
     * 2024-05-15T10:39:22.292
     */
    @Test
    public void msConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ms-consumer-group");

        consumer.setNamesrvAddr(MqConstant.NAMESRV_ADDR);

        consumer.subscribe("msTopic", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("receive message date: " + LocalDateTime.now());
                System.out.println("receive message: " + new String(list.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

        System.in.read();

    }


}
